/* 04-prod-cons

   Producer-consumer co-routining.  The producer puts 10 tokens into the 
   stream and the consumer consumes them.
*/
#include <pthread.h>
#include <stdlib.h>
#include <stdio.h>
#include <stdbool.h>
#include <signal.h>

/* One of these per Stream
   Holds:  the mutex lock and notifier condition variables
           a 'value' taken from a producer
           a variable 'available' indicating a token can be consumed
           an identity
*/
typedef struct {
   pthread_mutex_t lock;
   pthread_cond_t notifier;
   void *value;      /* object to be passed from consumer to producer */
   bool available;   /* 1 if there is an unread object to pass, otherwise 0 */
   int id;
} Stream;

/* return 'value' which should have a value from a call to put */
void *get(void *stream) {
   void *ret;

   bool available = ((Stream*)stream)->available;
   pthread_mutex_t *lock = &((Stream*)stream)->lock;
   pthread_cond_t *notifier = &((Stream*)stream)->notifier;
   
   pthread_mutex_lock(lock);    /* lock other threads out of this section    */
   if (!available)              /* if nothing in the buffer, wait and open   */
      pthread_cond_wait(notifier, lock);     /* the section to other threads */
   ((Stream*)stream)->available = false;     /* the token has been consumed  */
   ret = ((Stream*)stream)->value;           /* the token to be consumed     */
   pthread_cond_signal(notifier);  /* wake up sleeping producer, if any      */
   pthread_mutex_unlock(lock);  /* unlock the section */

   return ret;
}

/* 'value' is the value to move to the consumer */
void put(void *stream, void *value) {
   bool available = ((Stream*)stream)->available;
   pthread_mutex_t *lock = &((Stream*)stream)->lock;
   pthread_cond_t *notifier = &((Stream*)stream)->notifier;
   
   pthread_mutex_lock(lock);       /* lock the section */
   if (available)                  /* if a token is waiting for consumption  */
      pthread_cond_wait(notifier, lock);   /* put this thread to sleep       */
   ((Stream*)stream)->available = true;  /* a token is ready for consumption */
   ((Stream*)stream)->value = value;  /* this is the token to be consumed    */
   pthread_cond_signal(notifier);     /* wake up a sleeping consumer, if any */
   pthread_mutex_unlock(lock);     /* unlock the section */

   return;
}

/* Put 1,2,3,4,5... into the stream */
void *producer (void *stream) {
   int i;
   
   for (i=1 ; ; i++) {
      printf("Producer: sending %d\n",i);
      put(stream, (void*)i);
      printf("Producer: sent %d\n",i);
   }
   pthread_exit(NULL);
}

/* consume tokens from the producer's stream */
void *consumer (void *stream) {
   int i, value;

   for (i=0; i < 10; i++) printf("\t\t\tConsumer: got %d\n", (int)get(stream));

   pthread_exit(NULL);
}

/* initialize stream */
void init_stream (Stream *stream) {
   stream->available = false;
   stream->value = NULL;
   stream->id = 1;
}

int main () {
   pthread_t prod, cons;
   Stream stream;
   pthread_attr_t attr;

   /* initialize stream components */
   init_stream(&stream);

   /* initialize mutex and cond variables */
   pthread_mutex_init(&stream.lock, NULL);
   pthread_cond_init (&stream.notifier, NULL);

   pthread_attr_init(&attr);
   pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
   pthread_create(&prod, NULL, producer, (void*)&stream);
   pthread_create(&cons, NULL, consumer, (void*)&stream);
   
   pthread_join(cons, NULL);   /* cancel all running threads when the */
                               /* is finished consuming */
   pthread_cancel(prod);

   pthread_exit(NULL);
}
   
